/**
 * 上海中赢金融信息服务有限公司
 * Copyright (c) 2017-2027 Chinazyjr,Inc.All Rights Reserved.
 */
package com.sys.midware.hbase.hcore.Handler;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.Set;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Delete;
import org.apache.hadoop.hbase.client.Durability;
import org.apache.hadoop.hbase.client.HBaseAdmin;
import org.apache.hadoop.hbase.client.HConnection;
import org.apache.hadoop.hbase.client.HConnectionManager;
import org.apache.hadoop.hbase.client.HTableInterface;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.filter.Filter;
import org.apache.hadoop.hbase.io.compress.Compression.Algorithm;
import org.apache.hadoop.hbase.util.Bytes;

import com.sys.midware.hbase.hcore.HbaseConst;
import com.sys.midware.hbase.hcore.Optimize.ColumnFamilyOptimize;
import com.sys.midware.hbase.hcore.Optimize.HbaseMappingPool;
import com.sys.midware.hbase.hcore.Optimize.TableOptimize;
import com.sys.midware.hbase.hcore.query.HbaseQueryParam;
import com.sys.midware.hbase.hcore.query.QueryFactory;
import com.sys.midware.hbase.model.HbaseColumn;
import com.sys.midware.hbase.model.HbaseRow;
import com.sys.midware.logger.Logger;
import com.sys.midware.logger.LoggerFactory;

/**
 * 实现HBase表操作的类 <b>ClassName：</b> HBaseDBHandler <br/>
 * <b>Description：</b> TODO <br/>
 * <b>@author：</b> mobing <br/>
 * <b>@date：</b> 2015年10月28日 下午2:59:30 <br/>
 * <b>@version: </b> <br/>
 */
public class HBaseDBHandler implements DBHandler {

    private static Logger logger = LoggerFactory.getLogger(HBaseDBHandler.class);

    // define
    private Configuration HBASE_CONFIG = null;

    private String tableName = null;

    public HBaseDBHandler(final String tableName, Configuration HBASE_CONFIG) throws IOException {
        this.tableName = tableName;
        this.HBASE_CONFIG = HBASE_CONFIG;
    }

    public boolean isTableExists() throws Exception{
        boolean flag = false;
        HBaseAdmin admin = null;
        try {
            admin = new HBaseAdmin(HBASE_CONFIG);
            if (admin.tableExists(tableName)) {
                flag = true;
            }
        } catch (Exception e) {
            logger.error(e.getCause());
            throw e;
        } finally {
            admin.close();
        }
        return flag;
    }
    
    /**
     * 创建表
     * 
     * @param
     * @return void
     */
    private void perforCreate(Map<String, String> map) throws Exception {
        HBaseAdmin admin = null;
        try {
            String tableName = map.get(HbaseConst.TableAttr.KEY_TABLE_NAME);
            String cols[] = map.get(HbaseConst.TableAttr.KEY_COLUMN_FAMILY).split(",");
            // create table
            admin = new HBaseAdmin(HBASE_CONFIG);
            if (admin.tableExists(tableName)) {
                logger.debug("table has already existed:" + tableName);
                return;
            }
            TableName tableNameObj = TableName.valueOf(tableName);
            HTableDescriptor table = new HTableDescriptor(tableNameObj);
            for (String c : cols) {
                HColumnDescriptor col = new HColumnDescriptor(c);// column
                col.setMaxVersions(1);// default
                col.setCompressionType(Algorithm.GZ);// default
                table.addFamily(col);
            }
            admin.createTable(table);
            logger.debug("Table has been created success:" + tableName);
        } catch (Exception e) {
            logger.error(e.getCause());
            e.printStackTrace();
            throw e;
        } finally {
            admin.close();
        }

    }

    /**
     * creation with optimization arguments
     * 
     * @param hbaseTableOptimizeParam
     * @throws Exception
     */
    private void perforCreate(TableOptimize hbaseTableOptimizeParam) throws Exception {
        HBaseAdmin admin = null;
        try {
            String tableName = hbaseTableOptimizeParam.getTableName();
            // add to map
            HbaseMappingPool.map_tableOptimization.put(tableName, hbaseTableOptimizeParam);
            // create table
            admin = new HBaseAdmin(HBASE_CONFIG);

            // check
            if (admin.tableExists(tableName)) {
                logger.debug("table has already existed:" + tableName);
                return;
            }
            // to be created
            TableName tableNameObj = TableName.valueOf(tableName);
            HTableDescriptor table = new HTableDescriptor(tableNameObj);
            // set arguments
            if (hbaseTableOptimizeParam.isCompactionEnabled()) {
                table.setCompactionEnabled(hbaseTableOptimizeParam.isCompactionEnabled());
            }
            // set column family arguments
            for (ColumnFamilyOptimize cf : hbaseTableOptimizeParam.getColumnFamilys()) {
                // column
                HColumnDescriptor col = new HColumnDescriptor(cf.getColumnFamilyName());
                // set arguments
                col.setMaxVersions(cf.getMaxVersions());
                if (cf.getTimeToLive() != -1) {
                    col.setTimeToLive(cf.getTimeToLive());
                }
                if (cf.isInMemory()) {
                    col.setInMemory(cf.isInMemory());
                }
                col.setBloomFilterType(HbaseMappingPool.map_bloom.get(cf.getBloomType()));
                col.setCompressionType(HbaseMappingPool.map_compression.get(cf.getCompressionAlgorithm()));
                table.addFamily(col);
            }
            admin.createTable(table);
            logger.debug("Table has been created success:" + tableName);
        } catch (Exception e) {
            logger.error(e.getCause());
            e.printStackTrace();
            throw e;
        } finally {
            admin.close();
        }

    }

    /*
     * obj Map<String,String> <"tableName","Test"> <"cf","cf1,cf2"> or
     * HbaseTableOptimizeParam
     * 
     * @see com.b5m.spider.common.DBHandler#performCreate(java.lang.Object)
     */
    public void performCreate(Object obj) throws Exception {
        // check argument
        if (obj instanceof Map) {
            this.perforCreate((Map<String, String>) obj);
        } else if (obj instanceof TableOptimize) {
            this.perforCreate((TableOptimize) obj);
        } else {
            logger.error("arugument obj is not Map or HbaseTableOptimizeParam ", new IllegalArgumentException());
            throw new IllegalArgumentException(
                    "arugument obj is not Map or HbaseTableOptimizeParam :" + obj.getClass().getName());
        }
    }

    /*
     * obj :String :org.apache.hadoop.hbase.filter.Filter :List<Filter>
     * :HbaseQueryParam
     * 
     * @see com.b5m.spider.common.DBHandler#performQuery(java.lang.Object)
     */
    public List<Object> performQuery(Object obj) throws Exception {
        QueryFactory factory = null;
        List<Object> list = new ArrayList<Object>();
        HConnection connection = null;
        HTableInterface table = null;
        try {
            // create connection
            connection = HConnectionManager.createConnection(HBASE_CONFIG);
            table = connection.getTable(tableName);
            factory = new QueryFactory(table);
            if (obj instanceof byte[]) {
                Object temp = factory.queryByCondition((byte[]) obj);
                if (temp == null)
                    return Collections.emptyList();
                list.add(temp);
            } else if (obj instanceof HbaseQueryParam  && ((HbaseQueryParam)obj).getQueue() !=null) {//队列模式
                factory.queryByCondition4Queue((HbaseQueryParam) obj);
                return Collections.emptyList();
            } else if (obj instanceof Set) {
                list.addAll(factory.queryByCondition((Set<String>) obj));
            } else if (obj instanceof Filter) {
                list.addAll(factory.queryByCondition((Filter) obj));
            } else if (obj instanceof List) {
                list.addAll(factory.queryByCondition((List<Filter>) obj));
            } else if (obj instanceof HbaseQueryParam) {
                list.addAll(factory.queryByCondition((HbaseQueryParam) obj));
            } else {
                list = Collections.emptyList();
                logger.error("arugument obj is error", new IllegalArgumentException());
                throw new IllegalArgumentException("arugument obj is error :" + obj.getClass().getName());
            }
        } catch (Exception e) {
            logger.error(e);
            e.printStackTrace();
            throw e;
        } finally {
            this.close(connection, table);
        }
        return list;
    }
    
    /**
     * 根据rowkey和过滤条件查询
     * @param  
     * @return  List<Object>
     */
    public List<Object> performQuery(byte[] rowKey, Filter filter ) throws Exception {
        QueryFactory factory = null;
        List<Object> list = new ArrayList<Object>();
        HConnection connection = null;
        HTableInterface table = null;
        try {
            // create connection
            connection = HConnectionManager.createConnection(HBASE_CONFIG);
            table = connection.getTable(tableName);
            factory = new QueryFactory(table);
            Object temp =  factory.queryByCondition(rowKey, filter);
            if (temp == null)
                return Collections.emptyList();
            list.add(temp);
        } catch (Exception e) {
            logger.error(e);
            e.printStackTrace();
            throw e;
        } finally {
            this.close(connection, table);
        }
        return list;
    }

    public void performInsert(List<Object> records) throws Exception {
        HConnection connection = null;
        HTableInterface table = null;
        TableOptimize hbaseTableOptimizeParam = HbaseMappingPool.map_tableOptimization.get(tableName);
        try {
            // check
            if (records == null || records.isEmpty()) {
                throw new Exception("Insert-List is null or empty.");
            }
            Object e = records.get(0);
            if (!(e instanceof HbaseRow)) {
                logger.error("arugument obj is not List<HbaseRow>", new IllegalArgumentException());
                throw new IllegalArgumentException("arugument obj is not List<HbaseRow>:" + e.getClass().getName());
            }
            // create connection
            connection = HConnectionManager.createConnection(HBASE_CONFIG);
            table = connection.getTable(tableName);
            // set optimize args
            if (hbaseTableOptimizeParam != null && !hbaseTableOptimizeParam.isAutoFlushTo()) {
                table.setAutoFlushTo(hbaseTableOptimizeParam.isAutoFlushTo());//
            }
            // Sets the size of the buffer in bytes.
            if (hbaseTableOptimizeParam != null && Math.abs(hbaseTableOptimizeParam.getWriteBufferSize()) != -1) {
                table.setWriteBufferSize(hbaseTableOptimizeParam.getWriteBufferSize() * 1024 * 1024); // x
                                                                                                      // MB
            }
            // set default
            if (hbaseTableOptimizeParam == null) {
                table.setAutoFlushTo(false);
                table.setWriteBufferSize(5 * 1024 * 1024);// 5M
            }
            List<Put> puts = new ArrayList<Put>(records.size());
            // batch insert
            for (Object obj : records) {
                HbaseRow hbaseRow = (HbaseRow) obj;
                Put put = new Put(hbaseRow.getRow());
                // no write to WAL
                if (hbaseTableOptimizeParam != null) {
                    put.setDurability(HbaseMappingPool.map_wal.get(hbaseTableOptimizeParam.getWal()));
                }
                // set default
                if (hbaseTableOptimizeParam == null) {
                    put.setDurability(Durability.SKIP_WAL);
                }
                for (HbaseColumn hbaseColumn : hbaseRow.getColumns()) {
                    // 参数分别：列族、列、值
                    put.add(Bytes.toBytes(hbaseColumn.getColumnFamily()), Bytes.toBytes(hbaseColumn.getColumn()),
                            Bytes.toBytes(hbaseColumn.getValue()));
                }
                puts.add(put);
            }
            table.put(puts);// batch puts
            // table.flushCommits();//commit
        } catch (Exception e) {
            logger.error(e);
            e.printStackTrace();
            throw e;
        } finally {
            this.close(connection, table);
        }

    }

    @Override
    public void performInsert(Object record) throws Exception {
        if (record instanceof List) {
            this.performInsert((List) record);
            return;
        }
        List<Object> list = new ArrayList<Object>(1);
        list.add(record);
        this.performInsert(list);
    }

    @Override
    public void performUpdate(Object record) throws Exception {
        if (record instanceof List) {
            this.performUpdate((List) record);
            return;
        }
        this.performInsert(record);
    }

    @Override
    public void performDelete(Object record) throws Exception {
        if (record instanceof List) {
            this.performDelete((List) record);
            return;
        }
        List<Object> list = new ArrayList<Object>(1);
        list.add(record);
        this.performDelete(list);

    }

    @Override
    public void performUpdate(List<Object> records) throws Exception {
        this.performInsert(records);
    }

    /*
     * delete according to rowKeys
     * 
     * @see com.b5m.spider.common.DBHandler#performDelete(java.util.List)
     */
    public void performDelete(List<Object> records) throws Exception {
        HConnection connection = null;
        HTableInterface table = null;
        TableOptimize hbaseTableOptimizeParam = HbaseMappingPool.map_tableOptimization.get(tableName);
        try {
            // check
            if (records == null || records.isEmpty()) {
                throw new Exception("Delete List is null or empty.");
            }
            if (!(records.get(0) instanceof String) ) {
                logger.error("arugument obj is not List<String>", new IllegalArgumentException());
                throw new IllegalArgumentException(
                        "arugument obj is not List<String>:" + records.get(0).getClass().getName());
            }

            // create connection
            connection = HConnectionManager.createConnection(HBASE_CONFIG);
            table = connection.getTable(tableName);

            // set optimize args
            if (hbaseTableOptimizeParam != null && !hbaseTableOptimizeParam.isAutoFlushTo()) {
                table.setAutoFlushTo(hbaseTableOptimizeParam.isAutoFlushTo());//
            }
            // Sets the size of the buffer in bytes.
            if (hbaseTableOptimizeParam != null && Math.abs(hbaseTableOptimizeParam.getWriteBufferSize()) != -1) {
                table.setWriteBufferSize(hbaseTableOptimizeParam.getWriteBufferSize() * 1024 * 1024); // x
                                                                                                      // MB
            }

            // set default
            if (hbaseTableOptimizeParam == null) {
                table.setAutoFlushTo(false);
                table.setWriteBufferSize(5 * 1024 * 1024);// 5M
            }
            // delete
            List<Delete> list = new ArrayList<Delete>();
            for (Object rowKey : records) {
                Delete d = null;
                if(rowKey instanceof String){
                    d =  new Delete(Bytes.toBytes((String)rowKey));
                }
                // set default
                if (hbaseTableOptimizeParam == null) {
                    d.setDurability(Durability.SKIP_WAL);
                }
                list.add(d);
            }
            table.delete(list);// 删除
        } catch (Exception e) {
            logger.error(e);
            e.printStackTrace();
            throw e;
        } finally {
            this.close(connection, table);
        }

    }

    public void performDrop(Object obj) throws Exception {
        HBaseAdmin admin = null;
        try {
            // check argument
            if (!(obj instanceof String)) {
                logger.error("arugument obj is not String", new IllegalArgumentException());
                throw new IllegalArgumentException("arugument obj is not String :" + obj.getClass().getName());
            }

            // create table
            admin = new HBaseAdmin(HBASE_CONFIG);
            String tableName = (String) obj;

            if (admin.tableExists(tableName)) {
                admin.disableTable(tableName);
                admin.deleteTable(tableName);
                logger.debug("drop success,talbe:" + tableName);
            } else {
                logger.debug("table is not existed:" + tableName);
            }
            // remove from map
            HbaseMappingPool.map_tableOptimization.remove(tableName);
        } catch (Exception e) {
            logger.error(e.getCause());
            e.printStackTrace();
            throw e;
        } finally {
            admin.close();
        }
    }

    
  
    public void close() throws Exception {
    }

    private void close(HConnection connection, HTableInterface table) throws Exception {
        // close table
        if (table != null) {
            if (!connection.isClosed()) {
                table.close();
                connection.close();
            }
        }

    }

    @Override
    public boolean existed(Object obj) throws Exception {

        // TODO Auto-generated method stub
        return false;
    }
}
